package com.oig.sys.msg.thread;

import com.oig.sys.msg.app.IAppMqMsgService;
import com.oig.sys.msg.entity.TbMqMsg;
import com.oig.sys.msg.service.ITbMqConfigService;
import com.oig.sys.msg.vo.MqConfigVo;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;


/**
 * mq消息生产者
 * 轮询mq消息表，这是补发消息的，
 * 正常是往消息表存储消息事务提交后，立马发送然后删除消息，发送成功，删除不成功时有可能发送重复消息，消费端过滤
 */
@Slf4j
public class MqMsgPublisher {

    private final IAppMqMsgService appMqMsgService ;
    private final ITbMqConfigService mqConfigService;
    private final RabbitTemplate rabbitTemplate;

    public MqMsgPublisher(IAppMqMsgService appMqMsgService,
                          ITbMqConfigService mqConfigService,
                          RabbitTemplate rabbitTemplate){
        this.appMqMsgService = appMqMsgService;
        this.mqConfigService = mqConfigService;
        this.rabbitTemplate = rabbitTemplate;
    }


    private final ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
    private final Map<String, MqConfigVo> configCacheMap = new HashMap<>();   //简单缓存一下 只有一个线程调用，直接用hashMap

    public void start() {
        log.info("MqMsgPublisher Polling Thread start...");
        executor.scheduleWithFixedDelay(this::findAndReadyMsg, 0, 2, TimeUnit.SECONDS);
        executor.scheduleWithFixedDelay(this::findAndSendMqMsg, 0, 1, TimeUnit.SECONDS);
    }

    private void findAndReadyMsg(){
        //log.debug("mq消息发送准备轮询开始..");
        try{
            List<TbMqMsg> mqMsgList = this.appMqMsgService.findAllUnReadyMsg();  //一直在轮询，不会堆积很多消息没有发的。
            log.debug("mq消息发送准备查询结果：{}",mqMsgList==null?null:mqMsgList.size());
            if (mqMsgList!=null && !mqMsgList.isEmpty()){
                mqMsgList.forEach(this::dealReadMsg);
            }
        } catch (Exception ex){
            log.error("mq消息发送准备轮询..线程异常", ex);
        }
    }

    private void dealReadMsg(TbMqMsg mqMsg){
        try {
            //只能一个生效的mq配置
            MqConfigVo configVo = this.getMqConfig(mqMsg.getRelationType());
            if (configVo!=null){
                log.debug("mq消息发送准备..");
                this.appMqMsgService.updateMsgReady(mqMsg); //变成了，更新为待发送状态。
            } else {
                log.warn("mq消息发送准备 mq配置不存在..");
                this.appMqMsgService.updateMsgError(mqMsg) ;   //配置不存在标记一下异常，免得一直查到未发，实际上配置不正确，不会存消息
            }
        } catch (Exception e){
            log.error("mq发送准备处理出错",e);
        }
    }

    private void findAndSendMqMsg(){
        //log.debug("mq消息轮询开始..");
        try{
            List<TbMqMsg> mqMsgList = this.appMqMsgService.findAllUnSendMqMsg();  //一直在轮询，不会堆积很多消息没有发的。
            log.debug("mq消息轮询查询结果：{}",mqMsgList==null?null:mqMsgList.size());
            if (mqMsgList!=null && !mqMsgList.isEmpty()){
                mqMsgList.forEach(this::sendMqMsg);
            }
        } catch (Exception ex){
            log.error("mq消息轮询..线程异常",ex);
        }
    }

    private void sendMqMsg(TbMqMsg mqMsg){
        try {
            MqConfigVo configVo = this.getMqConfig(mqMsg.getRelationType());
            if (configVo==null){
                log.warn("发送配置没有");
                return;
            }
            log.debug("send msg:ex={},key={},msg={}",configVo.getProducerExchange(), configVo.getProducerKey(),mqMsg.getMsgContent());
            rabbitTemplate.convertAndSend(configVo.getProducerExchange(), configVo.getProducerKey(), mqMsg.getMsgContent());
            this.appMqMsgService.moveMsgToHis(mqMsg);   //发送成功后挪到历史表
        } catch (Exception e){
            log.error("mq消息发送处理出错",e);
            this.appMqMsgService.updateMsgError(mqMsg) ;
        }
    }

    private MqConfigVo getMqConfig(String cacheKey){
        MqConfigVo configVo = configCacheMap.get(cacheKey) ;
        if (configVo==null){
            log.debug("no cache...{}", cacheKey);
            configVo = this.mqConfigService.getOkMqConfig(cacheKey);
            if (configVo!=null){
                configCacheMap.put(cacheKey, configVo) ;
            }
        }
        return configVo;
    }


    public void toStop(){
        executor.shutdown();
    }

}
